using Confluent.Kafka;
using Glpt.Data;
using System;
using System.Collections.Generic;
using System.Data;

namespace Glpt.Task
{
    /// <summary>
    /// Kafka数据流类
    /// </summary>
    public class KafkaDataReader : IDataReader
    {
        private IConsumer<Ignore, string> consumer = null;
        private ConsumerConfig conf = null;
        private Dictionary<string, int> columnsByName = new Dictionary<string, int>();
        private Dictionary<int, string> columnsByOrdinal = new Dictionary<int, string>();
        private Dictionary<string, string> colnames_ext = new Dictionary<string, string>();//附加列
        private Dictionary<string, string> colnames_sep = new Dictionary<string, string>();//特殊附加列
        private int Cols_ext = 0; //附加列的个数
        private Dictionary<string, string> currentRow;
        private bool _isClosed = true;
        private bool cur_read = false;
        private int eofCount = 0;
        private int partition_count = 0;
        private string FIELD_PARTITION = "";
        private string FIELD_OFFSET = "";
        private bool isEnd = false;
        private bool AUTOCOMMIT = false;
        private TimeSpan tsp = TimeSpan.FromSeconds(10);
        private DateTime dt_end = DateTime.Now;
        /// <summary>
        /// 构造函数
        /// </summary>
        /// <param name="dbi">数据库连接</param>
        /// <param name="topic">表名</param>
        /// <param name="param">参数</param>
        public KafkaDataReader(DbItem dbi, string topic, Dictionary<string, string> param)
        {
            //全量读：将OFFSET设置为earliest，并且EnableAutoCommit = false，GROUPID可以固定使用一个
            //增量读：先将OFFSET设置为earliest，并且EnableAutoCommit = true,第二步再将OFFSET设置为Latest，GROUPID可以固定使用一个
            string BROKERLIST = dbi.DB_IP;
            string TOPIC = topic;
            if (param.ContainsKey("TOPIC") && param["TOPIC"].Length > 0)
                TOPIC = param["TOPIC"];
            string GROUPID = dbi.USER_NAME;
            if (param.ContainsKey("GROUPID") && param["GROUPID"].Length > 0)
                GROUPID = param["GROUPID"];
            AutoOffsetReset OFFSET_RESET = AutoOffsetReset.Earliest;
            if (param.ContainsKey("OFFSET_RESET") && param["OFFSET_RESET"].Length > 0)
                OFFSET_RESET = (AutoOffsetReset)int.Parse(param["OFFSET_RESET"]);
            if (param.ContainsKey("TIMEOUT") && param["TIMEOUT"].Length > 0)
                tsp = TimeSpan.FromSeconds(double.Parse(param["TIMEOUT"]));
            if (OFFSET_RESET.Equals(AutoOffsetReset.Earliest))
                dt_end = DateTime.Now.AddYears(1);
            else if (param.ContainsKey("EXEC_TIMEOUT") && param["EXEC_TIMEOUT"].Length > 0)
                dt_end = DateTime.Now.AddMinutes(double.Parse(param["EXEC_TIMEOUT"]));
            else
                dt_end = DateTime.Now.AddMinutes(60);
            if (param.ContainsKey("AUTOCOMMIT") && param["AUTOCOMMIT"].Length > 0)
                AUTOCOMMIT = param["AUTOCOMMIT"].Equals("1");

            foreach (string key in param.Keys)
            {
                if (key.StartsWith(":"))
                {
                    colnames_ext.Add(key.Substring(1), param[key]);
                    if (param[key].Equals("PARTITION"))
                        FIELD_PARTITION = key.Substring(1);
                    if (param[key].Equals("OFFSET"))
                        FIELD_OFFSET = key.Substring(1);
                }
            }
            conf = new ConsumerConfig
            {
                GroupId = GROUPID,
                BootstrapServers = BROKERLIST,
                AutoOffsetReset = OFFSET_RESET,
                EnablePartitionEof = true,
                EnableAutoCommit = AUTOCOMMIT
            };
            if (OFFSET_RESET.Equals(AutoOffsetReset.Latest))
            {
                conf.EnablePartitionEof = false;
                conf.EnableAutoCommit = true;
            }
            consumer = new ConsumerBuilder<Ignore, string>(conf)
            .SetPartitionsAssignedHandler((c, partitions) =>
            {
                partition_count = partitions.Count;
            })
            .Build();
            consumer.Subscribe(TOPIC);
            _isClosed = false;
            cur_read = Read();
            if (cur_read)
            {
                Cols_ext = 0;
                foreach (string key in colnames_ext.Keys)
                {
                    columnsByName.Add(key, Cols_ext);
                    columnsByOrdinal.Add(Cols_ext, key);
                    Cols_ext++;
                }
                int i = 0;
                foreach (string col_name in currentRow.Keys)
                {
                    columnsByName.Add(col_name, i + Cols_ext);
                    columnsByOrdinal.Add(i + Cols_ext, col_name);
                    i++;
                }
            }
        }
        public bool Read()
        {
            if (isEnd || DateTime.Now.CompareTo(dt_end) > 0)
                return false;
            if (cur_read)
            {
                cur_read = false;
            }
            else
            {
                var consumeResult = consumer.Consume(tsp);
                if (consumeResult == null)
                {
                    isEnd = true;
                    return false;
                }
                else if (consumeResult.IsPartitionEOF)
                {
                    eofCount++;
                    if (eofCount == partition_count)
                    {
                        isEnd = true;
                        return false;
                    }
                    else
                        return Read();
                }
                else
                {
                    if (consumeResult.Message == null || consumeResult.Message.Value == null)
                        return Read();
                    string message = consumeResult.Message.Value.Trim();
                    if (message.Length == 0 || message.Equals("{MSG_TOO_LONG}"))
                        return Read();
                    currentRow = JsonDataTable.JsonStringToDictionary(message);
                    if (currentRow.ContainsKey("ERROR") && currentRow["ERROR"].StartsWith("MSG_TOO_LONG"))
                        return Read();
                    if (FIELD_PARTITION.Length > 0)
                        colnames_ext[FIELD_PARTITION] = consumeResult.Partition.Value.ToString();
                    if (FIELD_OFFSET.Length > 0)
                        colnames_ext[FIELD_OFFSET] = consumeResult.Offset.Value.ToString();
                }
            }
            return true;
        }

        /// <summary>
        /// 关闭StreamReader对象和基础流
        /// </summary>
        public void Close()
        {
            if (consumer != null) consumer.Close();
            _isClosed = true;
        }

        public int FieldCount
        {
            get { return columnsByName.Count; }
        }
        public string GetString(int i)
        {
            if (i < Cols_ext)
                return colnames_ext[columnsByOrdinal[i]];
            else if (currentRow.ContainsKey(columnsByOrdinal[i]))
                return currentRow[columnsByOrdinal[i]];
            else
                return "";
        }

        public object GetValue(int i)
        {
            if (i < Cols_ext)
                return colnames_ext[columnsByOrdinal[i]];
            else if (currentRow.ContainsKey(columnsByOrdinal[i]))
                return currentRow[columnsByOrdinal[i]];
            else
                return DBNull.Value;
        }

        public string GetName(int i)
        {
            return columnsByOrdinal[i];
        }

        public int GetOrdinal(string name)
        {
            return columnsByName[name];
        }

        public int Depth
        {
            get
            {
                return 0;
            }
        }

        public bool IsClosed
        {
            get
            {
                return _isClosed;
            }
        }

        public int RecordsAffected
        {
            get
            {
                return -1;
            }
        }

        public DataTable GetSchemaTable()
        {
            DataTable dt = new DataTable();
            dt.Columns.Add("ColumnName", typeof(string));
            dt.Columns.Add("ColumnOrdinal", typeof(int));
            dt.Columns.Add("ColumnSize", typeof(int)).DefaultValue = 255;
            dt.Columns.Add("NumericPrecision", typeof(short)).DefaultValue = 255;
            dt.Columns.Add("NumericScale", typeof(short)).DefaultValue = 255;
            dt.Columns.Add("IsUnique", typeof(bool)).DefaultValue = false;
            dt.Columns.Add("IsKey", typeof(bool));
            dt.Columns.Add("BaseServerName", typeof(string));
            dt.Columns.Add("BaseCatalogName", typeof(string));
            dt.Columns.Add("BaseColumnName", typeof(string));
            dt.Columns.Add("BaseSchemaName", typeof(string));
            dt.Columns.Add("BaseTableName", typeof(string));
            dt.Columns.Add("DataType", typeof(System.Type));
            dt.Columns.Add("AllowDBNull", typeof(bool)).DefaultValue = false;
            dt.Columns.Add("ProviderType", typeof(int)).DefaultValue = 22;
            dt.Columns.Add("IsAliased", typeof(bool));
            dt.Columns.Add("IsExpression", typeof(bool));
            dt.Columns.Add("IsIdentity", typeof(bool)).DefaultValue = false;
            dt.Columns.Add("IsAutoIncrement", typeof(bool)).DefaultValue = false;
            dt.Columns.Add("IsRowVersion", typeof(bool)).DefaultValue = false;
            dt.Columns.Add("IsHidden", typeof(bool));
            dt.Columns.Add("IsLong", typeof(bool)).DefaultValue = false;
            dt.Columns.Add("IsReadOnly", typeof(bool)).DefaultValue = false;
            dt.Columns.Add("ProviderSpecificDataType", typeof(System.Type));
            dt.Columns.Add("DataTypeName", typeof(string)).DefaultValue = "string";
            dt.Columns.Add("XmlSchemaCollectionDatabase", typeof(string));
            dt.Columns.Add("XmlSchemaCollectionOwningSchema", typeof(string));
            dt.Columns.Add("XmlSchemaCollectionName", typeof(string));
            dt.Columns.Add("UdtAssemblyQualifiedName", typeof(string)).DefaultValue = 22;
            dt.Columns.Add("NonVersionedProviderType", typeof(int));
            dt.Columns.Add("IsColumnSet", typeof(bool)).DefaultValue = false;
            for (int i = 0; i < FieldCount; i++)
            {
                DataRow dr = dt.NewRow();
                dr["ColumnName"] = GetName(i);
                dr["ColumnOrdinal"] = i;
                dr["BaseColumnName"] = GetName(i);
                dr["DataType"] = typeof(string);
                dr["ProviderSpecificDataType"] = typeof(string);
                dt.Rows.Add(dr);
            }
            return dt;
        }

        public bool NextResult()
        {
            return false;
        }

        public object this[int i]
        {
            get
            {
                return GetValue(i);
            }
        }
        public object this[string name]
        {
            get
            {
                return GetValue(columnsByName[name]);
            }
        }
        public bool GetBoolean(int i)
        {
            return bool.Parse(GetString(i));
        }
        public byte GetByte(int i)
        {
            return byte.Parse(GetString(i));
        }
        public long GetBytes(int i, long fieldOffset, byte[] buffer, int bufferoffset, int length)
        {
            return -1;
        }
        public char GetChar(int i)
        {
            return char.Parse(GetString(i));
        }
        public long GetChars(int i, long fieldoffset, char[] buffer, int bufferoffset, int length)
        {
            return -1;
        }
        public IDataReader GetData(int i)
        {
            return null;
        }
        public string GetDataTypeName(int i)
        {
            return "String";
        }
        public DateTime GetDateTime(int i)
        {
            return DateTime.Parse(GetString(i));
        }
        public decimal GetDecimal(int i)
        {
            return decimal.Parse(GetString(i));
        }
        public double GetDouble(int i)
        {
            return double.Parse(GetString(i));
        }
        public Type GetFieldType(int i)
        {
            return typeof(string);
        }
        public float GetFloat(int i)
        {
            return float.Parse(GetString(i));
        }
        public Guid GetGuid(int i)
        {
            return Guid.Parse(GetString(i));
        }
        public short GetInt16(int i)
        {
            return short.Parse(GetString(i));
        }
        public int GetInt32(int i)
        {
            return int.Parse(GetString(i));
        }
        public long GetInt64(int i)
        {
            return long.Parse(GetString(i));
        }
        public int GetValues(object[] values)
        {
            for (int i = 0; i < columnsByName.Count; i++)
                values[i] = GetValue(i);
            return columnsByName.Count;
        }
        public bool IsDBNull(int i)
        {
            if (GetValue(i) == null)
                return true;
            else
                return false;
        }
        public void Dispose()
        {
            try
            {
                GC.Collect();
            }
            catch { }
        }
        ~KafkaDataReader()
        {
            Dispose();
        }
    }
}
